package com.example.hbasespringbootdemo.utils;

import com.spring4all.spring.boot.starter.hbase.api.HbaseTemplate;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Component
public class HbaseUtils {
    @Autowired
    private HbaseTemplate hbaseTemplate;

    public int createTable(String tableName,String[] family ) throws Exception{

        Admin admin = hbaseTemplate.getConnection().getAdmin();
        HTableDescriptor table = new HTableDescriptor(TableName.valueOf(tableName));
        for (String str : family) {
            HColumnDescriptor column = new HColumnDescriptor(str);
            table.addFamily(column);

        }

        if (admin.tableExists(TableName.valueOf(tableName))) {
            System.out.println("Table "+ tableName + "已经存在");
            return -1;
        }
        admin.createTable(table);
        admin.close();
        System.out.println(tableName+" is created");
        return 1;

    }

    public void insert(String tableName, String rowKey, String family, String column, String cell) throws Exception{
        Put put = new Put(rowKey.getBytes());
        Connection connection = hbaseTemplate.getConnection();
        if (checkTableName(tableName, connection)) {
            return;
        }
        Table table = connection.getTable(TableName.valueOf(tableName));
        put.add(family.getBytes(), column.getBytes(), cell.getBytes());
        table.put(put);
        table.close();
        System.out.println("insert data is ok");

    }

    private boolean checkTableName(String tableName, Connection connection) throws IOException {
        Admin admin = connection.getAdmin();
        if (!admin.tableExists(TableName.valueOf(tableName))) {
            System.out.println("the " + tableName + "is not exists");
            return true;
        }
        return false;
    }

    public void insertBranch(String tableName, String[] rowKey, String[] family, String[] column, String[] cell) throws Exception{
        Connection connection = hbaseTemplate.getConnection();
        Admin admin = connection.getAdmin();
        Table table = connection.getTable(TableName.valueOf(tableName));
        int length = rowKey.length;
        List<Put> putList = new ArrayList<Put>();
        if (!admin.tableExists(TableName.valueOf(tableName))) {
            System.out.println("the " + tableName + "is not exists");
            return;
        }
        for (int i =0; i < length; i++) {
            Put put = new Put(Bytes.toBytes(rowKey[i]));
            put.addColumn(Bytes.toBytes(family[i]), Bytes.toBytes(column[i]), Bytes.toBytes(cell[i]));
            putList.add(put);
        }
        table.put(putList);
        table.close();
        System.out.println("insert data is ok");

    }

    public Result queryByRow(String tableName, String rowKey) throws Exception {
        Get get = new Get(rowKey.getBytes());
        Connection connection = hbaseTemplate.getConnection();
        Table table = connection.getTable(TableName.valueOf(tableName));
        Result result = table.get(get);
        return result;
    }
}
